package com.xsomnus.rpc.consumer.netty;

import com.alibaba.fastjson.JSON;
import com.xsomnus.rpc.common.common.Constants;
import com.xsomnus.rpc.common.entity.Request;
import com.xsomnus.rpc.common.entity.Response;
import com.xsomnus.rpc.consumer.net.ConnectionManager;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.SynchronousQueue;

/**
 * @author @xsomnus666_xiawenye★
 * @since 2019/4/22 0022 16:47
 * 当你发现自己的才华撑不起野心时，就请安静下来学习吧
 */
@Slf4j
@Component
@ChannelHandler.Sharable
public class NettyClientHandler extends ChannelInboundHandlerAdapter {


    private final ConnectionManager connectionManager;

    private ConcurrentHashMap<String, SynchronousQueue<Object>> queueMap = new ConcurrentHashMap<>();


    public NettyClientHandler(ConnectionManager connectionManager) {
        this.connectionManager = connectionManager;
    }

    public SynchronousQueue<Object> sendRequest(Request request, Channel channel) {
        SynchronousQueue<Object> queue = new SynchronousQueue<>();
        queueMap.put(request.getId(), queue);
        channel.writeAndFlush(request);
        return queue;
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        log.debug("已连接到RPC服务器, {}", ctx.channel().remoteAddress());
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        log.debug("与RPC服务器断开连接: {}", ctx.channel().remoteAddress());
        ctx.channel().close();
        connectionManager.removeChannel(ctx.channel());
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        Response response = JSON.parseObject(msg.toString(), Response.class);
        String requestId = response.getRequestId();
        SynchronousQueue<Object> queue = queueMap.get(requestId);
        queue.put(response);
        queueMap.remove(requestId);
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        log.debug("已经超过30秒未与RPC服务器进行读写操作, 将发送心跳消息....");
        if (evt instanceof IdleStateEvent) {
            if (((IdleStateEvent) evt).state() == IdleState.ALL_IDLE) {
                Request request = new Request();
                request.setMethodName(Constants.HEART_BEAT_METHOD);
                ctx.writeAndFlush(request);
            }
        } else {
            super.userEventTriggered(ctx, evt);
        }

    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.debug("与RPC服务器通信发生异常: {}", cause.getMessage());
        ctx.channel().close();
    }
}
